package cn.xinfei.xdecision.pulsar.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Messages;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
@Slf4j
//@Service
public class XPulsarConsumerDemo {


    public XPulsarConsumerDemo() {
        log.info("---------------------------------");
    }

    @XPulsarConsumerAnnotate(
            topic = "xdecision-contenxt-snapshot",
//            topic = "xdecision-contenxt-snapshot",
            clazz = String.class,
            consumerName = "my-consumer",
            subscriptionName = "my-subscription",
            batch = true)
    public List<MessageId> consumeString(Messages<String> msgs) {

        try {
            if (null != msgs) {
                msgs.forEach((msg) -> {
                   log.info(new String(msg.getData()));
                    log.info(msg.getKey());
                    log.info(msg.getMessageId().toString());
                    log.info(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(msg.getEventTime())));
                    log.info(msg.getValue());
                    log.info(msg.getSequenceId()+"");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 推荐手动提交ack
     * @param msgs
     * @param consumer
     * @return
     */
    @XPulsarConsumerAnnotate(
            topic = "xdecision-contenxt-snapshot",
            clazz = String.class,
            consumerName = "my-consumer-1",
            subscriptionName = "my-subscription",
            batch = true)
    public List<MessageId> consumeStrings(Messages<String> msgs, Consumer<String> consumer) {
        try {
            if (null != msgs) {
                msgs.forEach((msg) -> {
                    log.info("---------------------------------------");

                    log.info(new String(msg.getData()));
                    log.info(msg.getKey());
                    MessageId messageId = msg.getMessageId();
                    log.info(messageId.toString());
                    log.info(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(msg.getEventTime())));
                    log.info(msg.getValue());
                    log.info(msg.getSequenceId()+"");
                    try {
                        log.info("ack messageId={}",messageId.toString());
                        consumer.acknowledgeAsync(messageId)
                                .handle((result, ex) -> {
                                    if (ex == null) {
                                        log.info("Message messageId: {} ", messageId);
//                                        successHandler.accept(messageId);
                                    } else {
                                        log.error("Error acking message to pulsar broker. errorMsg {}", ex);
//                                        failHandler.accept((Exception) ex);
                                    }
                                    return null;
                                });
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }
}
